package com.haotian.kafka;

import java.io.IOException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

/**
 * kafka 消费者 不可单利多线程调用应 每个线程创建一个消费者实例
 * @author 张昊天 2018年5月16日
 */
public class _KafkaConsumer extends KafkaBasic implements AutoCloseable {

	private KafkaConsumer<String, String> consumer;

	public _KafkaConsumer(_KafkaConfig config) throws UnknownHostException, IOException {
		super(config);
		consumer = KafkaFactory.createConsumer(config);
		consumer.subscribe(config.getTopics());
	}

	/**
	 * 消费者对象
	 * <p>
	 * </p>
	 * 
	 * @return
	 */
	public KafkaConsumer<String, String> getConsumer() {
		return this.consumer;
	}

	/**
	 * 消费一组数据 消费的数据会根据配置类中addTopic 来监听消费哪一个队列，根据配置类中setMaxPollRecords
	 * 来决定一次消费多少条； 本方法做了等待操作如果消费数据不存在 将1秒循环 直到等到数据 本方法为阻塞方法 放弃了 kafka本身自带的等待功能
	 * 会增大CPU使用量 和kafka的负载
	 * <p>
	 * </p>
	 * 
	 * @return
	 */
	public List<String> poll() {
		ConsumerRecords<String, String> poll = consumer.poll(0);
		while (poll.count() == 0) {
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// TODO 自动生成的 catch 块
				e.printStackTrace();
			}
			poll = consumer.poll(0);
		}
		List<String> values = new ArrayList<>();

		for (ConsumerRecord<String, String> consumerRecord : poll) {
			values.add(consumerRecord.value());
		}
		return values;
	}

	/**
	 * 关闭消费者
	 * <p>
	 * </p>
	 */
	public void close() {
		consumer.close();
	}
}
